package com.heartunderblade.rpc.DynamicProxy;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.heartunderblade.rpc.Registry.Registry;
import com.heartunderblade.rpc.RpcApplication;
import com.heartunderblade.rpc.RpcPojo.RpcRequest;
import com.heartunderblade.rpc.RpcPojo.RpcResponse;
import com.heartunderblade.rpc.RpcPojo.ServiceMetaInfo;
import com.heartunderblade.rpc.config.RpcConfig;
import com.heartunderblade.rpc.constant.RpcConstant;
import com.heartunderblade.rpc.factory.RegistryFactory;
import com.heartunderblade.rpc.factory.SerializerFactory;
import com.heartunderblade.rpc.handler.TcpServerHandler;
import com.heartunderblade.rpc.loadbalancer.LoadBalancer;
import com.heartunderblade.rpc.loadbalancer.LoadBalancerFactory;
import com.heartunderblade.rpc.protocol.*;
import com.heartunderblade.rpc.retry.RetryStrategy;
import com.heartunderblade.rpc.retry.RetryStrategyFactory;
import com.heartunderblade.rpc.service.Serializer;
import com.heartunderblade.rpc.tolerant.TolerantStrategy;
import com.heartunderblade.rpc.tolerant.TolerantStrategyFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.concurrent.Promise;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 自定义服务代理类：当用户调用某个接口方法时，会改为调用invoke方法，在invoke方法中获取到要调用的方法信息、参数列表等，通过这些参数构造请求参数进而完成调用
 */
public class DynamicProxy implements InvocationHandler {
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 指定序列化器
        //根据配置，获得序列化器实例，用于序列化请求数据
        Serializer serializer = SerializerFactory.getInstance(RpcApplication.getRpcConfig().getSerializer());
        // 构造请求
        RpcRequest rpcRequest = RpcRequest.builder()
                .ServiceName(method.getDeclaringClass().getName())
                .MethodName(method.getName())
                .ArgsType(method.getParameterTypes())
                .Args(args)
                .build();

        // 序列化
        byte[] bytes = null;
            bytes = serializer.serialize(rpcRequest);
            // 从注册中心获取服务提供者请求地址
            RpcConfig rpcConfig = RpcApplication.getRpcConfig();
            //根据工厂类和config得到注册中心
            Registry registry  = RegistryFactory.getInstance(rpcConfig.getRegistryConfig().getRegistry());
            ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
            serviceMetaInfo.setServiceName(method.getDeclaringClass().getName());
            serviceMetaInfo.setServiceVersion(RpcConstant.DEFAULT_SERVICE_VERSION);
            System.out.println(serviceMetaInfo);

            //拿到服务列表，serviceMetaInfo.getServiceKey()获取比如 服务名:版本号，再从registry.serviceDiscovery() 拿到对应的服务列表
            //         案例/rpc/myService:1.0/localhost:1234
            //可通过myService:1.0拿到所有/rpc/myService:1.0/，可知所有端口号的地址
            List<ServiceMetaInfo> serviceMetaInfoList = registry.serviceDiscovery(serviceMetaInfo.getServiceKey());
        System.out.println("动态代理"+serviceMetaInfoList);
            if (CollUtil.isEmpty(serviceMetaInfoList)) {
                throw new RuntimeException("暂无服务地址");
            }
            //使用负载均衡
            LoadBalancer loadBalancer  = LoadBalancerFactory.getInstance(rpcConfig.getLoadBalancer());
            // 将调用方法名（请求路径）作为负载均衡参数
            Map<String,Object> requestParams = new HashMap<>();
            requestParams.put("methodName",rpcRequest.getMethodName());
            ServiceMetaInfo select = loadBalancer.select(requestParams, serviceMetaInfoList);
            //扩展实现，使用重试机制发送,
            RpcResponse result = null;
            try {//加上容错机制
                RetryStrategy retryStrategy  = RetryStrategyFactory.getInstance(rpcConfig.getRetryStrategy());
                result = retryStrategy.doRetry(()->getRemoteRes(select.getServiceAddress(),rpcRequest));
            } catch (Exception e) {
                TolerantStrategy tolerantStrategy  = TolerantStrategyFactory.getInstance(rpcConfig.getTolerantStrategy());
                result = tolerantStrategy.doTolerant(null, e);
                throw new RuntimeException(e);
            }
            return  result.getData();
            // 发送请求

    }
    public static RpcResponse getRemoteRes(String url, RpcRequest rpcRequest) {
        // 构造协议
        ProtocolMessage<RpcRequest> protocolMessage = new ProtocolMessage<RpcRequest>();
        ProtocolMessage.Header header = new ProtocolMessage.Header();
        header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
        header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
        header.setSerializer((byte) ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer()).getKey());
        header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
        // 生成全局请求 ID
        header.setRequestId(IdUtil.getSnowflakeNextId());
        //设置他的BodyLength
        //获取序列化器
        Serializer instance = SerializerFactory.getInstance(RpcApplication.getRpcConfig().getSerializer());
        try {
            header.setBodyLength(instance.serialize(rpcRequest).length);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        protocolMessage.setHeader(header);
        protocolMessage.setBody(rpcRequest);
        System.out.println("动态代理的 message + " + protocolMessage);
        // 创建连接
        String host = url.split("://")[1].split(":")[0];
        int port = Integer.parseInt(url.split(":")[2]);
        EventLoopGroup group = new NioEventLoopGroup();

        // 使用一个Promise来保存返回的RpcResponse
        final RpcResponse[] result = new RpcResponse[1];

        try {
            // 创建一个事件循环组
            ChannelFuture future = new Bootstrap().group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new ProtocolMessageEncoderDecoder());
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO)); // 添加日志处理器
                            ch.pipeline().addLast(new SimpleChannelInboundHandler<ProtocolMessage>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext ctx, ProtocolMessage msg) throws Exception {
                                    // 在回调中保存结果
                                    result[0] = (RpcResponse) msg.getBody();
                                    // 通知Future，接收到响应
                                    ctx.channel().eventLoop().execute(() -> {
                                        synchronized (result) {
                                            result[0] = (RpcResponse) msg.getBody();
                                            result.notifyAll();
                                        }
                                    });
                                }

                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                                    cause.printStackTrace();
                                    ctx.close();
                                }
                            });
                        }
                    })
                    .connect(host, port).sync();

            // 向服务器发送消息
            future.channel().writeAndFlush(protocolMessage);

            // 等待直到收到响应，使用同步方法等待结果
            synchronized (result) {
                while (result[0] == null) {
                    result.wait();
                }
            }

            // 返回响应结果
            System.out.println("动态代理的 res  " + result[0]);
            return result[0];
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        } finally {
            // 释放线程池资源
            group.shutdownGracefully();
        }
    }

}
